package org.mq.mymq.nameserver.replication;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.internal.StringUtil;
import org.mq.mymq.common.coder.TcpMsgDecoder;
import org.mq.mymq.common.coder.TcpMsgEncoder;
import org.mq.mymq.common.utils.AssertUtil;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.common.MasterSlaveReplicationProperties;
import org.mq.mymq.nameserver.common.NameServerProperties;
import org.mq.mymq.nameserver.common.TraceReplicationProperties;
import org.mq.mymq.nameserver.enums.ReplicationModeEnum;
import org.mq.mymq.nameserver.enums.ReplicationRoleEnum;
import org.mq.mymq.common.event.EventBus;
import org.mq.mymq.nameserver.handler.MasterReplicationServerHandler;
import org.mq.mymq.nameserver.handler.NodeSendMsgReplicationServerHandler;
import org.mq.mymq.nameserver.handler.NodeWriteMsgReplicationServerHandler;
import org.mq.mymq.nameserver.handler.SlaveReplicationServerHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @ClassName ReplicationService
 * @Description 提供服务注册信息分片复制存储服务
 * @Author jiarong_ye
 * @Date 2025/6/19 23:47
 * @Version 1.0
 */
public class ReplicationService {
    private static final Logger log = LoggerFactory.getLogger(ReplicationService.class);

    // 参数校验
    public ReplicationModeEnum checkProperties() {
        NameServerProperties props = CommonCache.getNameserverProperties();
        String mode = props.getReplicationMode();
        if (StringUtil.isNullOrEmpty(mode)) {
            log.info("单机模式下运行注册中心");
            return null;
        }
        // 为空参数不合法抛异常
        ReplicationModeEnum replicationModeEnum = ReplicationModeEnum.of(mode);
        AssertUtil.isNotNull(replicationModeEnum, "复制模式参数异常");
        if (replicationModeEnum == ReplicationModeEnum.TRACE) {
            // 链路复制
            TraceReplicationProperties traceReplicationProperties = props.getTraceReplicationProperties();
            AssertUtil.isNotNull(traceReplicationProperties.getPort(), "node节点端口为空");
        } else {
            // 主从复制
            MasterSlaveReplicationProperties masterSlaveReplicationProperties = props.getMasterSlaveReplicationProperties();
            AssertUtil.isNotBlank(masterSlaveReplicationProperties.getMaster(), "master参数为空");
            AssertUtil.isNotBlank(masterSlaveReplicationProperties.getRole(), "role参数为空");
            AssertUtil.isNotBlank(masterSlaveReplicationProperties.getType(), "type参数为空");
            AssertUtil.isNotNull(masterSlaveReplicationProperties.getPort(), "同步端口参数为空");
        }
        return replicationModeEnum;
    }
    // 根据参数判断复制方式
    public void  startReplicationTask(ReplicationModeEnum replicationModeEnum) {
        if (replicationModeEnum == null) {
            // 单机版本
            return;
        }
        int port = 0;
        NameServerProperties nameserverProperties = CommonCache.getNameserverProperties();
        if (replicationModeEnum == ReplicationModeEnum.MASTER_SLAVE) {
            port = nameserverProperties.getMasterSlaveReplicationProperties().getPort();
        } else {
            replicationModeEnum = ReplicationModeEnum.TRACE;
        }
        ReplicationRoleEnum roleEnum;
        if (replicationModeEnum == ReplicationModeEnum.MASTER_SLAVE) {
            roleEnum = ReplicationRoleEnum.of(nameserverProperties.getMasterSlaveReplicationProperties().getRole());
        } else {
            String nextNode = nameserverProperties.getTraceReplicationProperties().getNextNode();
            if (StringUtil.isNullOrEmpty(nextNode)) {
                roleEnum = ReplicationRoleEnum.TAIL_NODE;
            } else {
                roleEnum = ReplicationRoleEnum.NODE;
            }
            port = nameserverProperties.getTraceReplicationProperties().getPort();
        }
        int replicationPort = port;
        if (roleEnum == ReplicationRoleEnum.MASTER) {
            startNettyServerAsync(new MasterReplicationServerHandler(new EventBus("master-replication-task-")), replicationPort);
        } else if (roleEnum == ReplicationRoleEnum.SLAVE) {
            String masterAddress = nameserverProperties.getMasterSlaveReplicationProperties().getMaster();
            startNettyConnAsync(new SlaveReplicationServerHandler(new EventBus("slave-replication-task-")), masterAddress);
        } else if (roleEnum == ReplicationRoleEnum.NODE) {
            String nextNodeAddress = nameserverProperties.getTraceReplicationProperties().getNextNode();
            startNettyServerAsync(new NodeWriteMsgReplicationServerHandler(new EventBus("node-write-msg-replication-task-")), replicationPort);
            startNettyConnAsync(new NodeSendMsgReplicationServerHandler(new EventBus("node-send-msg-replication-task-")), nextNodeAddress);
        } else if (roleEnum == ReplicationRoleEnum.TAIL_NODE) {
            startNettyServerAsync(new NodeWriteMsgReplicationServerHandler(new EventBus("node-write-msg-replication-task-")), replicationPort);
        }
    }
    // 启动netty进程/单机模式不启动


    /**
     * 开启一个主从架构下从节点的链接服务
     *
     * @param simpleChannelInboundHandler
     * @param address ip:port
     */
    private void startNettyConnAsync(SimpleChannelInboundHandler simpleChannelInboundHandler, String address) {
        Thread nettyConnThread = new Thread(() -> {
            EventLoopGroup clientEventLoopGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(clientEventLoopGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TcpMsgDecoder());
                    ch.pipeline().addLast(new TcpMsgEncoder());
                    ch.pipeline().addLast(simpleChannelInboundHandler);
                }
            });
            ChannelFuture channelFuture = null;
            try {
                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                    clientEventLoopGroup.shutdownGracefully();
                    log.info("nameserver' replication connect application is closed");
                }));
                String[] addr = address.split(":");
                channelFuture = bootstrap.connect(addr[0], Integer.valueOf(addr[1])).sync();
                // 链接了master节点的对象，需要保存做数据交互
                Channel ch = channelFuture.channel();
                CommonCache.setConnectNodeChannel(ch);
                log.info("success connected to nameserver replication {}", address);
                ch.closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        nettyConnThread.setName("netty-conn-replication-thread");
        nettyConnThread.start();
    }

//     开启一个netty服务
//     master-slave架构下传入的handler分为：
//     master的handler，监听验证从节点链接事件，链接成功后将从节点channel注册到分片复制管理器中，后续同步数据用
//     slave的handler，发送链接事件到主节点，然后接受主节点同步数据（验证不通过，要重试）
//
//     trace架构下传入的handler分为：
//     主节点handler：监听验证从节点链接事件，链接成功后将从节点channel注册到分片复制管理器中，后续同步数据用
//     从节点handler：发送链接事件到上一个节点，并等待下一个节点链接事件，接收前节点数据发送给下一个节点


    /**
     * 开启一个主从架构下主节点服务
     *
     * @param simpleChannelInboundHandler
     */
    private void startNettyServerAsync (SimpleChannelInboundHandler simpleChannelInboundHandler, int port) {
        Thread nettyServerTask = new Thread(() -> {
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            // 处理网络IO中的read&write事件
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup);
            bootstrap.channel(NioServerSocketChannel.class);
            bootstrap.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new TcpMsgDecoder());
                    ch.pipeline().addLast(new TcpMsgEncoder());
                    ch.pipeline().addLast(simpleChannelInboundHandler);
                }
            });
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
                log.info("nameserver's replication application is closed");
            }));
            ChannelFuture channelFuture = null;
            try {
                // master-slave架构下
                // master节点会开启一个写入数据到从节点的服务
                // slave节点会发送链接消息链接主节点
                // trace架构下，需要接收外界数据也要写给下一个节点
                channelFuture = bootstrap.bind(port).sync();
                log.info("success connected to nameserver replication {}", port);
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        nettyServerTask.setName("netty-server-replication-task");
        nettyServerTask.start();
    }

}
